package org.zjt.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Description:
 *
 *   命令启动:
 *   $ /bin/flink run -c org.myorg.quickstart.SocketTextStreamWordCount /your/path/target/original-flink-quickstart-java-1.0-SNAPSHOT.jar 127.0.0.1 9000
 * @author juntao.zhang
 * Date: 2018-09-29 上午10:41
 * @see
 */
public class SocketTextStreamWordCount {


    public static void main(String[] args) throws Exception {
        //参数检查
        String hostName = "127.0.0.1";
        Integer port = 9000;

        //设置环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据
        DataStream<String> text = env.socketTextStream(hostName, port);

        //计数   keyBy position   sum position
        DataStream<Tuple2<String, Integer>> counts = text.flatMap(new LineSplitter()).keyBy(0).sum(1);
        counts.print();

        // 提交执行
        env.execute("Java WordCount from SocketTextStream Example");
    }


    /**
     * 行的单词拆分
     */
    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {

            String[] tokens = s.toLowerCase().split("\\W+");

            for (String token: tokens) {
                if (token.length() > 0) {
                    collector.collect(new Tuple2(token, 1));
                }
            }
        }
    }
}

